package com.atguigu.api3

import java.util

import com.atguigu.api.SensorReading
import org.apache.flink.api.common.functions.{RichMapFunction, RichReduceFunction}
import org.apache.flink.api.common.state._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._

/**
 *
 * @description: 在processTime处理模式下,对延时数据可以直接将时间延时若干单位时间,
 *               在eventTime处理模式下,对延时数据就可以通过watermark机制来处理,watermark用于表示timestamp小于Watermark的数据都已经达到.用来让程序自己平衡延迟和结果正确性
 *               watermark是一条特殊的数据记录
 *               必须单调递增,以确保任务的事件时间时钟都在向前推进,而不是在后退,watermark与数据的时间戳相关,
 *               首先确定延时3s[数据中最大差值],0-5s的窗口 6-10 首先数据都各自放到相应
 *               的桶里面,然后来了8 8-3=5那么0-5这个窗口就可以关闭计算了,后来的4则丢弃,有了watermark后延时处理数据关闭窗口也是以watermark为基准的.如210-225 225-245 来了一个285s的数据那么当前watermark为285-3=282,
 *               此时282距离225s为57s还达不到关窗的条件,如果来到了289s数据进来了.那么watermark=286 286-225=61延时也满足了直接关窗聚合计算了.
 * @time: 2021-03-15 15:35
 * @author: baojinlong 
 **/
object F02StateTestDemo {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行度
    environment.setParallelism(1)
    val inputStream: DataStream[String] = environment.socketTextStream("localhost", 7777)
    val dataStream: DataStream[SensorReading] = inputStream.map {
      data => {
        val arr: Array[String] = data.split(",")
        SensorReading(arr(0), timestamp = arr(1).toLong, temperature = arr(2).toDouble)
      }
    }


    environment.execute("state test")
  }

}

// 泛型表示输入输出,同时keyState必须定义在RichFunction中,因为需要运行时上下文
class MyRichMapper extends RichMapFunction[SensorReading, String] {
  var valueState: ValueState[Double] = _
  // 第二种定义方式
  lazy val listState: ListState[Int] = getRuntimeContext.getListState(new ListStateDescriptor("listStateTest", classOf[Int]))
  lazy val mapState: MapState[String, Double] = getRuntimeContext.getMapState(new MapStateDescriptor[String, Double]("mapStateTest", classOf[String], classOf[Double]))
  // 聚合完成后的值SensorReading
  lazy val reduceState: ReducingState[SensorReading] = getRuntimeContext.getReducingState(new ReducingStateDescriptor[SensorReading]("reduceStateTest", new MyDiyReducerDemo, classOf[SensorReading]))

  override def open(parameters: Configuration): Unit = {
    valueState = getRuntimeContext.getState(new ValueStateDescriptor[Double]("valueStateTest", classOf[Double]))
  }

  override def map(in: SensorReading): String = {
    val myValue: Double = valueState.value
    println("获取到键控状态的值为" + myValue)
    valueState.update(in.temperature)

    // listState 操作api
    val ints = new util.ArrayList[Int]
    ints.add(1)
    ints.add(2)
    listState.addAll(ints)
    listState.update(ints)
    listState.get()

    mapState.contains("hello")
    mapState.put("sensor", 11)

    val nowReduceValue: SensorReading = reduceState.get
    println(s"当前聚合结果为$nowReduceValue")
    // 再跟之前的结果进行聚合
    reduceState.add(in)
    in.id
  }
}

class MyDiyReducer extends RichReduceFunction[SensorReading] {
  override def reduce(t: SensorReading, t1: SensorReading): SensorReading = {
    SensorReading(t1.id, Math.min(t.temperature, t1.temperature).toLong, t1.timestamp)
  }
}